package com.ucex.common.rocketmq.consumer.ali;

import java.util.Properties;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.ucex.common.rocketmq.MQProperties;
import com.ucex.common.rocketmq.consumer.AbstractMessageConsumerProcess;
import com.ucex.common.rocketmq.consumer.MessageConsumer;

import lombok.extern.slf4j.Slf4j;

/**
 * 阿里MQ消费者
 *
 */
@Slf4j
public class AliMQConsumerProcess extends AbstractMessageConsumerProcess {

    private static Consumer consumer;

    private static volatile AliMQConsumerProcess INSTANCE;

    private AliMQConsumerProcess() {}

    public static AliMQConsumerProcess getInstance() {
        if (INSTANCE == null) {
            synchronized (AliMQConsumerProcess.class) {
                if (INSTANCE == null) {
                    INSTANCE = new AliMQConsumerProcess();
                }
            }
        }
        return INSTANCE;
    }

    /**
     * 初始化AliMQ配置信息
     */
    @Override
    public void initConsumer(MQProperties mqProperties) {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.AccessKey, mqProperties.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, mqProperties.getSecretKey());
        properties.put(PropertyKeyConst.NAMESRV_ADDR, mqProperties.getNameSrvAddr());
        properties.put(PropertyKeyConst.GROUP_ID, mqProperties.getGroupId());
        consumer = ONSFactory.createConsumer(properties);
    }

    @Override
    public void handlerMessage(final MessageConsumer listener) {
        consumer.subscribe(listener.getTopic(), listener.getTag(), new MessageListener() {
            public Action consume(Message message, ConsumeContext consumeContext) {
                String tag = message.getTag();
                if (null == tag || tag.trim().isEmpty()) {
                    log.info("Wrong message, 'tag' is null or empty, message: {}", message);
                    return Action.CommitMessage;
                }
                try {
                    String msg = new String(message.getBody());
                    if (msg != null && !msg.trim().isEmpty()) {
                        listener.onMessage(msg);
                    }
                    return Action.CommitMessage;
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    return Action.ReconsumeLater;
                }
            }
        });
    }

    @Override
    public void start() {
        consumer.start();
    }
}
